package com.codejiwei.flink.table;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.MapTypeInfo;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @author jiwei
 * @description
 * @date 2023/8/8 8:36
 */
public class Flink_Table_API_06 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);


        //主流数据
        tEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "  `user_id` BIGINT,\n" +
                "  `item_id` BIGINT,\n" +
                "  `behavior` STRING,\n" +
                "  `ts` TIMESTAMP(3) \n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'drity_data_source',\n" +   //
                "  'properties.bootstrap.servers' = '192.168.102.154:29092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +  //'earliest-offset'，'latest-offset'，'group-offsets'，'timestamp' 和 'specific-offsets'
                "  'json.ignore-parse-errors' = 'true',\n" +
                "  'format' = 'json'\n" +
                ")");

        Table sourceTable = tEnv.from("sourceTable");
        DataStream<Row> sourceDS = tEnv.toDataStream(sourceTable);


        //规则流数据
        SingleOutputStreamOperator<Map<String, Long>> ruleSource = env.readTextFile("src/main/resources/data/rule_test.csv")
                .map(new MapFunction<String, Map<String, Long>>() {
                    @Override
                    public Map<String, Long> map(String value) throws Exception {
                        String[] fields = value.split(",");
                        HashMap<String, Long> map = new HashMap<>();
                        map.put(fields[0], Long.parseLong(fields[1]));
                        return map;
                    }
                }
    );

        //定义广播state
        MapStateDescriptor<String, Map<String, Long>> RULE_BROADCAST_CONFIG = new MapStateDescriptor<>(
                "RuleBroadcast",
                BasicTypeInfo.STRING_TYPE_INFO,
                new MapTypeInfo<>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO)
        );

        BroadcastStream<Map<String, Long>> ruleDS = ruleSource.broadcast(RULE_BROADCAST_CONFIG);

        SingleOutputStreamOperator<String> processDS = sourceDS.connect(ruleDS)
                .process(new BroadcastProcessFunction<Row, Map<String, Long>, String>() {
                    @Override
                    public void processElement(Row value, BroadcastProcessFunction<Row, Map<String, Long>, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        Map<String, Long> ruleBroadcast = ctx.getBroadcastState(RULE_BROADCAST_CONFIG).get("RuleBroadcast");

                        Long userId = value.<Long>getFieldAs("user_id");
                        Long rule = ruleBroadcast.get("user_id");

                        if (userId > rule) {
                            out.collect(value.toString());
                        }
                    }

                    @Override
                    public void processBroadcastElement(Map<String, Long> value, BroadcastProcessFunction<Row, Map<String, Long>, String>.Context ctx, Collector<String> out) throws Exception {
                        ctx.getBroadcastState(RULE_BROADCAST_CONFIG).put("RuleBroadcast", value);
                    }
                });

        tEnv.createTemporaryView("SinkTable", processDS);


        env.execute();

    }
}
